package com.xdu.yunwufog.connection.netty;

import com.xdu.yunwufog.common.exception.ServerErrorException;
import com.xdu.yunwufog.connection.utils.ConnectionUtil;
import com.xdu.yunwufog.connection.utils.DataBusConstant;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.AbstractChannelPoolMap;
import io.netty.channel.pool.ChannelPoolMap;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;


import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * netty连接池
 */
public class NettyClientPool {
    // volatile用来确保将变量的更新操作通知到其他线程。
    volatile private static NettyClientPool nettyClientPool;
    private final Logger logger = LoggerFactory.getLogger(getClass());
    /**
     * key为目标主机的InetSocketAddress对象，value为目标主机对应的连接池
     * InetSocketAddress可以为ip+port,也可以为hostname+port
     * FixedChannelPool:ChannelPool，可以强制保持一个最大的连接并发
     */

    final EventLoopGroup group = new NioEventLoopGroup();
//    final EventLoopGroup boss = new NioEventLoopGroup();
    final Bootstrap bootstrap = new Bootstrap();
    // 注意这里的pools是private，poolMap应该是用来跟pools进行信息更新
    volatile private static Map<InetSocketAddress,FixedChannelPool> pools = new HashMap<>(16);
    // 注意此处使用了volatile 进行了隔离
    volatile private static List<InetSocketAddress> addressList;

    private NettyClientPool(){
        build();
    }

    /**
     * @Author voidwx
     * @Description 获取线程池的单例
     * @Date 17:05 2022/4/13
     * @Param []
     * @return com.xdu.yunwufog.connection.netty.NettyClientPool
     **/
    public static NettyClientPool getInstance(){
        if (nettyClientPool == null){
            // 同步操作，即加锁
            synchronized (NettyClientPool.class){
                // 为了避免多次初始化，此处又重新做了一次null值判断
                if (nettyClientPool == null){
                    nettyClientPool = new NettyClientPool();
                }
            }
        }
        return nettyClientPool;
    }

    public void build(){
        // 应该使用日志
        // System.out.println("NettyClientPool build...");
        logger.info("NettyClientPool build...");
        bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true)
                .option(ChannelOption.SO_KEEPALIVE,true)
                .option(ChannelOption.RCVBUF_ALLOCATOR,new AdaptiveRecvByteBufAllocator(1024*2,1024*10,1024*1024));
        // 添加了这一行，将channel缓存动态扩容

        // 为对应的IP+PORT创建channelPool
        for(int i = 0; i < ConnectionUtil.CHANNEL_POOL_CONN_NUM.length; i++){
            InetSocketAddress address = ConnectionUtil.ADDRESSES[i];
            int num = ConnectionUtil.CHANNEL_POOL_CONN_NUM[i];
            pools.put(ConnectionUtil.ADDRESSES[i],new FixedChannelPool(bootstrap.remoteAddress(address),new NettyChannelPoolHandler(),num,num));
        }

    }
    /**
     * 功能描述：
     *  根据随机数取出的server对应pool，从pool中取出channel
     *  连接池的动态扩容： 指定最大连接数为Integer.MAX_VALUE,如果连接池队列中取不到channel，会自动创建channel，默认使用FIFO的获取方式，回收的channel优先被再次get到
     *  SERVER的宕机自动切换: 指定重试次数，get()发生连接异常，会重新获取
     * @param retry 可重试的次数

     */

    public Channel getChannel(long random, InetSocketAddress address, int retry) throws ServerErrorException{
//        int retry = 0;
        Channel channel = null;
        try {
            // random是一个关于时间戳的随机数
            // 根据address获取对应的连接池
            FixedChannelPool pool = pools.get(address);
            // 从连接池获取连接
            Future<Channel> future = pool.acquire();
            channel = future.get();
            // 为channel设置key:random随机数
            AttributeKey<Long> randomID = AttributeKey.valueOf(DataBusConstant.RANDOM_KEY);
            channel.attr(randomID).set(random);
        } catch (ExecutionException e) {  //如果是因为服务端挂掉，连接失败而获取不到channel，则随机数执行+1操作，从下一个池获取
            logger.info(e.getMessage());
            // 每个池子尝试获取2次
            if(retry > 0){
                return getChannel( random,address,retry - 1);
            } else {
                logger.error("Peer Server address {} Error",address);
                throw new ServerErrorException("Server Error");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return channel;
    }

    /**
     *  回收channel进池，需要保证随机值和getChannel获取到的随机值是同一个，才能从同一个pool中释放资源
     * @param ch
     */
    public static void release(Channel ch,InetSocketAddress socketAddress){
       long random = ch.attr(AttributeKey.<Long>valueOf(DataBusConstant.RANDOM_KEY)).get();
       ch.flush();
       pools.get(socketAddress).release(ch);
    }

    /**
     * 获取线程池的hash值
     */
    public static int getPoolHash(Channel ch){
        // 获取random随机值
        long random = ch.attr(AttributeKey.<Long>valueOf(DataBusConstant.RANDOM_KEY)).get();
        InetSocketAddress address = (InetSocketAddress) ch.remoteAddress();
        return System.identityHashCode(pools.get(address));
    }

    /**
     * 获取服务端server列表，每个server对应一个pool
     *
     */
    @Deprecated
    public void getInetAddresses(String addresses){
        addressList = new ArrayList<>(3);
        // 此处需要注意看是否会出错
        if (StringUtil.isNullOrEmpty(addresses)){
            throw new RuntimeException("address列表为空");
        }
        String[] splits = addresses.split(",");
        for (String address : splits){
            String[] split = address.split(":");
            if (split.length==0){
                throw new RuntimeException("["+address+"]不符合IP:PORT格式");
            }
            addressList.add(new InetSocketAddress(split[0],Integer.parseInt(split[1])));
        }
    }
}
